package org.ricks.net.gossip;

import java.io.IOException;
import java.net.*;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

public class SimpleGossip {
    private String LOCAL_IP = "127.0.0.1";
    private int PORT = 6000;
    private static final int GOSSIP_INTERVAL = 3000;
    private static final int DIRECT_TIMEOUT = 1000;
    private static final int INDIRECT_TIMEOUT = 2000;

    enum NodeStatus { ALIVE, SUSPECT, DEAD }

    static class NodeState {
        String nodeId;
        InetSocketAddress address;
        NodeStatus status;
        long version;
        long lastUpdate;

        // ✨ 新增toString()方法用于序列化
        @Override
        public String toString() {
            return nodeId + ":" + status + ":" + version;
        }
    }

    private final Map<String, NodeState> clusterNodes = new ConcurrentHashMap<>();
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
    private final DatagramSocket socket;

    public SimpleGossip(String local,String seedNodes) throws Exception {
        String[] str = local.split(":");
        this.LOCAL_IP = str[0];
        this.PORT = Integer.valueOf(str[1]);

        this.socket = new DatagramSocket(PORT);
        initSeedNodes(seedNodes);
        startGossipProcess();
        startUdpListener();
        System.out.println("Gossip节点启动，端口：" + PORT); // ✨ 添加启动日志
    }

    private void initSeedNodes(String seeds) {
        for (String addr : seeds.split(",")) {
            String[] parts = addr.split(":");    // ✨ 修复数组越界问题
            InetSocketAddress address = new InetSocketAddress(parts[0], Integer.parseInt(parts[1]));
            NodeState node = new NodeState();
            node.nodeId = addr;
            node.address = address;
            node.status = NodeStatus.ALIVE;
            node.version = System.currentTimeMillis();
            node.lastUpdate = System.currentTimeMillis();
            clusterNodes.put(addr, node);
        }
    }

    private void startGossipProcess() {
        scheduler.scheduleAtFixedRate(() -> {
            List<NodeState> targets = selectRandomNodes(3);
            System.out.println("本轮探测节点：" + targets.stream()
                    .map(n -> n.nodeId)
                    .collect(Collectors.joining(", "))); // ✨ 添加调试日志
            for (NodeState target : targets) {
                sendPing(target.address);
            }
        }, 0, GOSSIP_INTERVAL, TimeUnit.MILLISECONDS);
    }

    private void startUdpListener() {
        new Thread(() -> {
            byte[] buffer = new byte[1024];
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
                    socket.receive(packet);
                    String message = new String(packet.getData(), 0, packet.getLength());
                    System.out.println("收到消息：" + message); // ✨ 添加接收日志
                    processMessage(message);
                    // 原有监听逻辑
                } catch (ArrayIndexOutOfBoundsException e) { // ✨ 捕获特定异常
                    System.err.println("消息格式异常：" + e.getMessage());
                } catch (Exception e) {
                    System.err.println("监听异常：" + e.getClass().getSimpleName());
                }
            }
        }).start();
    }

    private void sendPing(InetSocketAddress target) {
        try {
            String msg = "PING:" + LOCAL_IP + ":" + PORT;
            byte[] data = msg.getBytes();
            DatagramPacket packet = new DatagramPacket(data, data.length, target);
            socket.send(packet);

            // ✨ 新增间接探测逻辑
            scheduler.schedule(() -> {
                List<NodeState> proxies = selectRandomNodes(2);
                proxies.forEach(proxy ->
                        sendMessage(proxy.address, "INDIRECT_PING:" +target.getHostName() + ":"+ target.getPort())
                );
                scheduler.schedule(() -> checkResponse(target), INDIRECT_TIMEOUT, TimeUnit.MILLISECONDS);
            }, DIRECT_TIMEOUT, TimeUnit.MILLISECONDS);

        } catch (Exception e) {
            System.err.println("发送PING失败：" + e.getMessage());
        }
    }

    // ✨ 补全消息处理逻辑
    private void processMessage(String message) {
        String[] parts = message.split(":");
        if (parts.length < 2) {
            System.err.println("非法消息格式：" + message);
            return;
        }

        String msgType = parts[0];
        String payload = parts[1];
        String port = parts[2];

        switch (msgType) {// ✨ 修正为parts[0]判断消息类型
        case "PING":
            handlePing(parseAddress(payload + ":" + port));
             break;
        case "ACK":
            updateNodeStatus(parseNodeId(payload + ":" + port), NodeStatus.ALIVE);
            break;
        case "STATUS":
        mergeNodeState(payload + ":" + port, Long.parseLong(parts[4]), NodeStatus.valueOf(parts[3]));
        break;
        case "INDIRECT_PING":
        handleIndirectPing(port);
        break;
        }
    }

    // ✨ 新增辅助方法
    private String parseNodeId(String addrStr) {
        String[] parts = addrStr.split(":");
        return parts[0] + ":" + parts[1];  // host:port格式
    }

    private InetSocketAddress parseAddress(String addrStr) {
        String[] parts = addrStr.split(":");
        return new InetSocketAddress(parts[0], Integer.parseInt(parts[1]));
    }

    private synchronized void updateNodeStatus(String nodeId, NodeStatus newStatus) {
        NodeState node = clusterNodes.get(nodeId);
        if (node != null) {
            // ✨ 仅当状态真正改变时才更新
            if (node.status != newStatus) {
                long newVersion = System.currentTimeMillis();
                node.status = newStatus;
                node.version = newVersion;
                node.lastUpdate = newVersion;

                System.out.println("更新节点状态：" + node.nodeId + " -> " + newStatus);
                broadcastStatus(node);
            }
        } else {
            // ✨ 新增节点自动发现逻辑（当收到未知节点的ACK时）
            System.out.println("发现新节点：" + nodeId);
            String[] parts = nodeId.split(":");
            InetSocketAddress address = new InetSocketAddress(parts[0], Integer.parseInt(parts[1]));
            NodeState newNode = new NodeState();
            newNode.nodeId = nodeId;
            newNode.address = address;
            newNode.status = newStatus;
            newNode.version = System.currentTimeMillis();
            clusterNodes.put(nodeId, newNode);
        }
    }

    // ✨ 新增方法相关调用点增强
    private void handlePing(InetSocketAddress sender) {
        // ✨ 增加发送者状态同步逻辑
        String senderId = sender.getHostName() + ":" + sender.getPort();
        updateNodeStatus(senderId, NodeStatus.ALIVE);

        sendMessage(sender, "ACK:" + LOCAL_IP+":" + PORT);
    }
    // ✨ 补全handleIndirectPing方法
    private void handleIndirectPing(String targetPort) {
        int port = Integer.parseInt(targetPort);
        clusterNodes.values().stream()
                .filter(n -> n.address.getPort() == port)
                .findFirst()
                .ifPresent(node ->
                        sendMessage(node.address, "ACK:" + LOCAL_IP+":" + PORT)
                );
    }

    // ✨ 补全mergeNodeState方法
    private synchronized void mergeNodeState(String nodeId, long version, NodeStatus status) {
        NodeState local = clusterNodes.get(nodeId);
        if (local == null) {
            local = new NodeState();
            local.nodeId = nodeId;
            local.version = version - 1;
        }

        if (version > local.version) {
            local.status = status;
            local.version = version;
            local.lastUpdate = System.currentTimeMillis();
            System.out.println("合并节点状态：" + local); // ✨ 添加状态变更日志
        }
    }

//    // ✨ 补全sendMessage方法
//    private void sendMessage(InetSocketAddress target, String message) {
//        try {
//            byte[] data = message.getBytes();
//            DatagramPacket packet = new DatagramPacket(data, data.length, target);
//            socket.send(packet);
//        } catch (Exception e) {
//            System.err.println("发送消息失败：" + e.getMessage());
//        }
//    }

    // ✨ 重构checkResponse方法
    private void checkResponse(InetSocketAddress target) {
        NodeState node = clusterNodes.values().stream()
                .filter(n -> n.address.equals(target))
                .findFirst()
                .orElse(null);

        if (node != null && node.status == NodeStatus.ALIVE) {
            System.out.println("节点 " + node.nodeId + " 响应超时");
            markAsSuspect(node);
        }
    }

    private void broadcastStatus(NodeState changedNode) {
        // 构建状态消息：类型:节点ID:状态:版本号
        String statusMsg = String.format("STATUS:%s:%s:%d",
                changedNode.nodeId,
                changedNode.status.name(),
                changedNode.version);

        // 获取当前节点的端口用于自我过滤
        final int selfPort = socket.getLocalPort();

        // 线程安全遍历节点集合副本
        new ArrayList<>(clusterNodes.values()).forEach(node -> {
            // 避免给自己发送广播（根据端口判断）
            if (node.address.getPort() != selfPort) {
                System.out.println("广播状态到 " + node.nodeId + ": " + statusMsg);
                sendMessage(node.address, statusMsg);
            }
        });
    }

    private void sendMessage(InetSocketAddress target, String message) {
        try {
            byte[] data = message.getBytes(StandardCharsets.UTF_8);
            DatagramPacket packet = new DatagramPacket(data, data.length, target);
            socket.send(packet);
        } catch (IOException e) {
            System.err.println("消息发送失败至 " + target + ": " + e.getMessage());
            // 自动触发故障检测逻辑
            if(target.getPort() != PORT) {  // 排除种子节点
                clusterNodes.values().stream()
                        .filter(n -> n.address.equals(target))
                        .findFirst()
                        .ifPresent(this::markAsSuspect);
            }
        }
    }


    // ✨ 优化markAsSuspect方法
    private void markAsSuspect(NodeState node) {
        node.status = NodeStatus.SUSPECT;
        node.version = System.currentTimeMillis();
        System.out.println("标记为可疑：" + node.nodeId);
        broadcastStatus(node);

        scheduler.schedule(() -> {
            if (node.status == NodeStatus.SUSPECT) {
                node.status = NodeStatus.DEAD;
                node.version = System.currentTimeMillis();
                System.out.println("标记为死亡：" + node.nodeId);
                broadcastStatus(node);
                clusterNodes.remove(node.nodeId); // ✨ 死亡节点清理
            }
        }, 5000, TimeUnit.MILLISECONDS);
    }

    private List<NodeState> selectRandomNodes(int count) {
        List<NodeState> nodes = new ArrayList<>(clusterNodes.values());
        Collections.shuffle(nodes);
        return nodes.subList(0, Math.min(count, nodes.size()));
    }

    public static void main(String[] args) throws Exception {
//        if (args.length < 1) {
//            System.err.println("用法：java SimpleGossip <端口> [种子节点]");
//            return;
//        }
//        int port = Integer.parseInt(args[0]);
//        String seeds = args.length > 1 ? args[1] : "";
        new SimpleGossip("127.0.0.1:5002","127.0.0.1:5000,127.0.0.1:5001"); // ✨ 修改构造方法支持动态端口
    }
}
